package com.demo.hbase.utils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;

public class HBASEUtil {

	private Connection connection = null;

	public static String ROW_KEY = "rowkey";

	public HBASEUtil(String zkclientPort, String zkquorum, String hbmaster) {
		Configuration configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.property.clientPort", zkclientPort);
		configuration.set("hbase.zookeeper.quorum", zkquorum);
		configuration.set("hbase.master", hbmaster);
		try {
			connection = ConnectionFactory.createConnection(configuration);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public Connection getHbaseConnection() throws IOException {
		return connection;
	}

	/**
	 * 创建表
	 * 
	 * @param tableName
	 * @throws IOException
	 */
	public void createTable(String tableName, String... columns)
			throws IOException {
		if (columns == null || columns.length < 1) {
			throw new Error("columns不能为空");
		}
		Admin hBaseAdmin = connection.getAdmin();
		TableName tname = TableName.valueOf(tableName);
		// 如果存在要创建的表，那么先删除，再创建
		if (hBaseAdmin.tableExists(tname)) {
			hBaseAdmin.disableTable(tname);
			hBaseAdmin.deleteTable(tname);
		}
		HTableDescriptor tableDescriptor = new HTableDescriptor(tname);
		for (String column : columns) {
			tableDescriptor.addFamily(new HColumnDescriptor(column));
		}
		hBaseAdmin.createTable(tableDescriptor);
	}

	/**
	 * 插入数据
	 * 
	 * @param tableName
	 * @throws IOException
	 */
	public void insertData(String tableName, List<Map<String, byte[]>> datas)
			throws IOException {
		if (datas == null || datas.size() < 1)
			return;
		Table table = connection.getTable(TableName.valueOf(tableName));
		List<Put> puts = new ArrayList<Put>(datas.size());
		for (Map<String, byte[]> row : datas) {
			if (row == null || row.isEmpty())
				continue;
			byte[] rowkey = row.get(ROW_KEY);
			if (rowkey == null) {
				rowkey = UUID.randomUUID().toString().getBytes();
			}
			Put put = new Put(rowkey);
			for (Map.Entry<String, byte[]> entry : row.entrySet()) {
				String key = entry.getKey();
				if (ROW_KEY.equals(key))
					continue;
				byte[] value = entry.getValue();
				put.addColumn(key.getBytes(), null, value);
			}
			puts.add(put);
		}
		table.put(puts);
	}

	/**
	 * 删除一张表
	 * 
	 * @param tableName
	 * @throws IOException
	 */
	public void dropTable(String tableName) throws IOException {
		Admin hBaseAdmin = connection.getAdmin();
		TableName tname = TableName.valueOf(tableName);
		hBaseAdmin.disableTable(tname);
		hBaseAdmin.deleteTable(tname);
	}

	/**
	 * 根据 rowkey删除一条记录
	 * 
	 * @param tableName
	 * @param rowkey
	 * @throws IOException
	 */
	public void deleteRecordsByRowkey(String tableName, String... rowkey)
			throws IOException {
		if (rowkey == null || rowkey.length < 1)
			return;
		TableName tname = TableName.valueOf(tableName);
		Table table = connection.getTable(tname);
		List<Delete> list = new ArrayList<Delete>();
		for (String key : rowkey) {
			Delete d1 = new Delete(key.getBytes());
			list.add(d1);
		}
		table.delete(list);
	}
	
	/**
	 * 单条件查询,根据rowkey查询唯一一条记录
	 * @param tableName
	 * @param rowkey
	 * @return
	 * @throws IOException
	 */
	public Cell[] getRecordByRowkey(String tableName, String rowkey) throws IOException{
		Get scan = new Get(rowkey.getBytes());//根据rowkey查询 
		TableName tname = TableName.valueOf(tableName);
		Table table = connection.getTable(tname);
        Result r = table.get(scan);  
		return r.rawCells();
	}
	
	/**
	 * 
	 * @param tableName
	 * @param filters
	 * @return
	 * @throws IOException
	 */
	public ResultScanner getResultScanner(String tableName, List<Filter> filters) throws IOException{
		FilterList filterList1 = new FilterList(filters);  
        Scan scan = new Scan();  
        scan.setFilter(filterList1);
        TableName tname = TableName.valueOf(tableName);
        Table table = connection.getTable(tname);
		return table.getScanner(scan);
	}
	
}
